package cn.ilikes.tools.hbase.jpa.clients.datahander;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.slf4j.Logger;

import cn.ilikes.tools.hbase.jpa.JPAHbasePersistenUnitEnv;
import cn.ilikes.tools.hbase.jpa.SysConfigs;
import cn.ilikes.tools.hbase.jpa.clients.datasource.HbaseDataSource;
import cn.ilikes.tools.hbase.jpa.clients.nodes.Node;
import cn.ilikes.tools.hbase.jpa.clients.nodes.Node.NodeLink;
import cn.ilikes.tools.hbase.jpa.metadatas.IEntityMetaData;
import cn.ilikes.tools.hbase.jpa.metadatas.elements.EntityFields;
import cn.ilikes.tools.hbase.jpa.metadatas.elements.Relation;
import cn.ilikes.tools.logs.Loggerfactory;
import cn.ilikes.tools.untils.HBaseUtils;
import cn.ilikes.tools.untils.PropertyAccessorHelper;

public class HbaseDataHander extends AbDataHander {

	private Logger logger = Loggerfactory.instance(HbaseDataHander.class);

	public HbaseDataHander(HbaseDataSource dataSource, JPAHbasePersistenUnitEnv hbasePersistenUnitEnv) {
		super(dataSource, hbasePersistenUnitEnv);

	}

	private HTableInterface getHbaseHtable(IEntityMetaData entityMetaData) {
		HbaseDataSource hbaseDataSource = getDataSource();
		String entityTableName = entityMetaData.getEntityTableName();
		HTableInterface hTableInterface = hbaseDataSource.findHTable(entityTableName);
		return hTableInterface;
	}

	@Override
	public void remove(Node node) throws JPAHBaseDataHnaderException {

		IEntityMetaData entityMetaData = node.getEntityMetaData();
		Object entity = node.getEntity();

		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "remove  entity  : " + entity);
			HTableInterface hTableInterface = getHbaseHtable(entityMetaData);
			byte[] idKey = PropertyAccessorHelper.getObjectToByte(entity, entityMetaData.getEntityId().getField());
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "remove  entity id : " + PropertyAccessorHelper.fieldValue(entity, entityMetaData.getEntityId().getField()));
			Delete delete = new Delete(idKey);
			hTableInterface.delete(delete);

		} catch (IOException e) {
			throw new JPAHBaseDataHnaderException(e);
		}

	}

	@Override
	public Result find(IEntityMetaData entityMetaData, Object key) throws JPAHBaseDataHnaderException {
		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "find  entity   key : " + key + "  entity class : " + entityMetaData.getEntityClazz());
			HTableInterface hTableInterface = getHbaseHtable(entityMetaData);
			Get get = new Get(HBaseUtils.getBytes(key));
			Result result = hTableInterface.get(get);
			return result;
		} catch (IOException e) {
			throw new JPAHBaseDataHnaderException(e);
		}
	}

	//EntityFieldsType RealocalKey   Relation  : true   jPAField is  ColumnName  :  false     columnName  :    null     Relation  ColumnName  :  PHOTO_ID
	//EntityFieldsType RealocalKey   Relation  : false   jPAField is  ColumnName  :  true     columnName  :    ALBUM_NAME     Relation  ColumnName  :  null
	//EntityFieldsType RoreignKey   Relation  : true   jPAField is  ColumnName  :  false     columnName  :    null     Relation  ColumnName  :  ALBUM_ID
	@Override
	public void save(Node node) throws JPAHBaseDataHnaderException {
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, "save  entity  : " + node.getEntity());
		HTableInterface hTableInterface = getHbaseHtable(node.getEntityMetaData());
		byte[] idKey = PropertyAccessorHelper.getObjectToByte(node.getEntity(), node.getEntityMetaData().getEntityId().getField());
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, "save  entity id : " + PropertyAccessorHelper.fieldValue(node.getEntity(), node.getEntityMetaData().getEntityId().getField()));
		try {
			Put put = new Put(idKey);
			Set<Map.Entry<String, EntityFields>> entityFields = node.getEntityMetaData().getEntityFields().entrySet();
			for (Entry<String, EntityFields> entityField : entityFields) {
				{
					EntityFields entityFields_ = entityField.getValue();
					if (entityFields_.getjPAField() == null ||entityFields_.isSave()==false)continue;
					switch (entityFields_.getEntityFieldsType()) {
					case RealocalKey: {
						if (entityFields_.getjPAField().isRelation()) {//有关系
							//有节点链接类型
							if (node.getNodeLink().size() > 0) {
								NodeLink nodeLink = node.getNodeLink().get(entityFields_);
								if (nodeLink == null)
									continue;
								Object value = nodeLink.getTargetObject();
								if (value == null)
									continue;
								Relation relation = entityFields_.getjPAField().getRelation();
								Class<?> targetEntity = relation.getTargetEntity();
								IEntityMetaData targetEntityMetaData = getHbasePersistenUnitEnv().getEntityMetaDataFactory().selectIEntityMetaData(
										targetEntity);
								EntityFields targetEntityFieldsid = targetEntityMetaData.getEntityId();

								Object id = PropertyAccessorHelper.getObject(value, targetEntityFieldsid.getField());
								String columnName = relation.getColumnName();
								int index = columnName.indexOf(SysConfigs.hbase_column_family_qualifier);
								if (Loggerfactory.isdebug(logger))
									Loggerfactory.debug(logger, "save entity realocal key relation column name  :  " + columnName + "  value : "
											+ id);
								put.add(HBaseUtils.getBytes(HBaseUtils.getfamily(columnName, index)),
										HBaseUtils.getBytes(HBaseUtils.getQualifier(columnName, index)), HBaseUtils.getBytes(id));

							}

						} else {//无关系
							
							Object value = PropertyAccessorHelper.getObject(node.getEntity(), entityFields_.getField());
							if (value == null)continue;
							if (Loggerfactory.isdebug(logger))
								Loggerfactory.debug(logger, "save  entity  field  column  name  :  " + entityFields_.getjPAField().getColumnName()
										+ "  value : " + value + "            entityFields  " + entityFields_.getFieldName());
							String columnName = entityFields_.getjPAField().getColumnName();
							int index = columnName.indexOf(SysConfigs.hbase_column_family_qualifier);
							put.add(HBaseUtils.getBytes(HBaseUtils.getfamily(columnName, index)),
									HBaseUtils.getBytes(HBaseUtils.getQualifier(columnName, index)), HBaseUtils.getBytes(value));
						}

						break;
					}

					case RoreignKey: {
						Relation relation = entityFields_.getjPAField().getRelation();
						Class<?> targetEntity = relation.getTargetEntity();
						IEntityMetaData targetEntityMetaData = getHbasePersistenUnitEnv().getEntityMetaDataFactory().selectIEntityMetaData(targetEntity);
						NodeLink nodeLink = node.getNodeLink().get(entityFields_);
						Object id = PropertyAccessorHelper.getObject(nodeLink.getTargetObject(), targetEntityMetaData.getEntityId().getField());
						String columnName = relation.getColumnName();
						int index = columnName.indexOf(SysConfigs.hbase_column_family_qualifier);
						if (Loggerfactory.isdebug(logger))
							Loggerfactory.debug(logger, "save entity roreign key relation column name  :  " + columnName + "  value : " + id);
						put.add(HBaseUtils.getBytes(HBaseUtils.getfamily(columnName, index)), HBaseUtils.getBytes(HBaseUtils.getQualifier(columnName, index)),
								HBaseUtils.getBytes(id));
					}
					default:
					}
				}
			}
			hTableInterface.put(put);
		} catch (IOException e) {
			throw new JPAHBaseDataHnaderException(e);
		}
	}

	@Override
	public Result[] seach(IEntityMetaData entityMetaData, List keyIds) throws JPAHBaseDataHnaderException {
		try {
			String entityTableName = entityMetaData.getEntityTableName();
			HbaseDataSource hbaseDataSource = getDataSource();
			HTableInterface hTableInterface = hbaseDataSource.findHTable(entityTableName);
			List<Get> gets = new ArrayList<Get>();
			for (Object keyId : keyIds) {
				Get get = new Get(HBaseUtils.getBytes(keyId));
				gets.add(get);
			}
			Result[] results = hTableInterface.get(gets);
			return results;
		} catch (IOException e) {
			throw new JPAHBaseDataHnaderException(e);
		}

	}

	/**
	 * 重写
	 */
	@Override
	public void update(Node node) throws JPAHBaseDataHnaderException {
		this.remove(node);
		this.save(node);

	}

	@Override
	public byte[] findValue(IEntityMetaData entityMetaData,Object key, String family, String  qualifier) throws JPAHBaseDataHnaderException {
		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "find  Value  : " + key + "  family : " + family+"    qualifier "+qualifier);
			HTableInterface hTableInterface = getHbaseHtable(entityMetaData);
			Get get = new Get(HBaseUtils.getBytes(key));
			Result result = hTableInterface.get(get);
			return result.getValue(HBaseUtils.getBytes(family),HBaseUtils.getBytes( qualifier));
		} catch (IOException e) {
			throw new JPAHBaseDataHnaderException(e);
		}
	}
}